package com.shux.kafka.consumer;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 描述：发送数据到客户端，客户端接收，并且设定异常策略
 * 
 * <pre>HISTORY
 * ****************************************************************************
 *  ID   DATE           PERSON          REASON
 *  1    2017年9月20日      Simba.Hua         Create
 * ****************************************************************************
 * </pre>
 * @author Simba.Hua
 */
public class ConsumerMessageSender<T> {
	private Logger logger = LoggerFactory.getLogger(getClass());
	
	private ExecutorService executorService;
	private ConsumerConfig consumerConfig;
	private ConsumerOptionConfig optionConfig;
	private BlockingQueue<KafkaBatchMessage<T>> packMessageQueue;
	private IMessageListener<T> messageListener;
	private ShuxKafkaConsumer<T> kafkaConsumer;
	
	public ConsumerMessageSender(ConsumerOptionConfig optionConfig,BlockingQueue<KafkaBatchMessage<T>> packMessageQueue,ShuxKafkaConsumer<T> kafkaConsumer,IMessageListener<T> messageListener) {
		this.optionConfig = optionConfig;
		this.packMessageQueue = packMessageQueue;
		this.kafkaConsumer = kafkaConsumer;
		this.messageListener = messageListener;
		executorService = Executors.newFixedThreadPool(optionConfig.getConsumerThreadCount(), new ThreadFactory(){
			int index = 1;
			@Override
			public Thread newThread(Runnable r) {
				Thread t = new Thread(r,"consumer-sender-"+ ++index);
				t.setDaemon(true);
				return t;
			}
		});
	}
	public void sendConsumerMessage(){
		for(int i = 0; i < optionConfig.getConsumerThreadCount(); i++ ){
			executorService.submit(() -> {
				for(;;){//不断从queue中取数据
					KafkaBatchMessage<T> batchMessage = null;
					int calcount = 0;
					try {
						logger.info("------------------packMessageQueue size" + packMessageQueue.size());
						batchMessage = packMessageQueue.take();
						logger.info("从packMessageQueue获取数据，并发送给Listener:"+batchMessage);
						for(;;){//如果出错还要继续发送;
							try {
								messageListener.onMessage(batchMessage.getMessages());
								break;
							} catch (Throwable e) {
								if(optionConfig.getExceptionHandler().retry(e, calcount)) {//继续尝试发送消息
									String msg = String.format("consumer error,will retry immediately,calcount=%d,topic=%s,message=%s", new Object[]{calcount,consumerConfig.getTopic(),batchMessage});
									logger.warn(msg,e);
								} else {
									throw e;//退出
								}
							}
						}
						calcount ++;
					} catch (InterruptedException e) {
						logger.warn("take kafkaBatchMessage from packMessageQueue is interrupted",e);
						Thread.currentThread().interrupt();
					} catch(Throwable t) {
						String errorMsg = String.format("consumer error,calcount=%d,topic=%s,message=%s", new Object[]{calcount,consumerConfig.getTopic(),batchMessage});
						logger.error(errorMsg,t);
						throw t;
					} finally {
						kafkaConsumer.completeMessage(batchMessage);
					}
				}
			});
		}
		
	}
	public ExecutorService getExecutorService() {
		return executorService;
	}
	public void setExecutorService(ExecutorService executorService) {
		this.executorService = executorService;
	}
	
	
}
